from kafka import KafkaConsumer
import mysql.connector
import re
from datetime import datetime

# MySQL配置
db_config = {
    'user': 'root',
    'password': 'Abcd@1234',
    'host': 'localhost',
    'database': 'data_shixun'
}

# Kafka配置
KAFKA_TOPIC = 'test'
KAFKA_SERVER = 'localhost:9092'

# 创建MySQL连接
cnx = mysql.connector.connect(**db_config)
cursor = cnx.cursor()

# 创建Kafka消费者
consumer = KafkaConsumer(KAFKA_TOPIC, bootstrap_servers=[KAFKA_SERVER])

# 正则表达式用于解析日志
log_pattern = re.compile(r'''
    (?P<user_id>\S+)\s+
    (\S+)\s+
    (\S+)\s+
    \[(?P<timestamp>.*?)\]\s+
    "(.*?)"\s+
    (?P<system>\S+)\s+
    (?P<source>\S+)\s+
    (?P<fans>(NO|YES))\s+
    (?P<opus_id>\S+)\s+
    (?P<thumb>(NO|YES))\s+
    (?P<collection>(NO|YES))\s+
    (?P<comment>(NO|YES))\s+
    (?P<finash>(NO|YES))
''', re.VERBOSE)
for message in consumer:
    log_line = message.value.decode('utf-8').strip()
    match = re.match(log_pattern, log_line)

    if match:
        user_id = match.group('user_id')
        timestamp = match.group('timestamp')
        system = match.group('system')
        source = match.group('source')
        fans = match.group('fans')
        opus_id = match.group('opus_id')
        thumb = match.group('thumb')
        collection = match.group('collection')
        comment = match.group('comment')
        finash = match.group('finash')

        # 插入到MySQL
        insert_query = """
        INSERT INTO data_user_behavior (user_id, timestamp, system, source, fans, opus_id, thumb, collection, comment, finash)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        """
        cursor.execute(insert_query, (user_id, timestamp, system, source, fans, opus_id, thumb, collection, comment, finash))
        cnx.commit()

# 关闭连接
cursor.close()
cnx.close()